package com.wngbms.websocket.zookeeper;

import com.wngbms.websocket.user.UserHook;
import com.wngbms.websocket.support.SpringBeanSupport;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.*;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;

import java.time.LocalDateTime;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent.Type;
import static org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent.Type.*;

@Slf4j
public class CuratorStartRunner implements ApplicationRunner, ZkConst {

    CuratorFramework curatorFramework;
    ZookeeperSupport zookeeperSupport;
    ExecutorService zookeeperThreadPool;

    public CuratorStartRunner(CuratorFramework curatorFramework, ZookeeperSupport zookeeperSupport, ExecutorService zookeeperThreadPool) {
        this.curatorFramework = curatorFramework;
        this.zookeeperSupport = zookeeperSupport;
        this.zookeeperThreadPool = zookeeperThreadPool;
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        try {
            log.info("connecting curator client...");
            curatorFramework.start();
            boolean connected = curatorFramework.blockUntilConnected(30, TimeUnit.SECONDS);
            if (!connected) {
                throw new IllegalStateException("zookeeper not connected");
            }

            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                // 删除服务节点
                zookeeperSupport.deletePath(getServiceRootPath());

                log.info("close curator client...");
                curatorFramework.close();
            }, "curator-close-thread"));

            // 创建坐席服务总根节点
            log.info("create zookeeper root node for seat-service...");
            if (!zookeeperSupport.checkExists(NODE_ROOT_PATH)) {
                log.info("zookeeper root node '/rx-seat' not exist...");
                zookeeperSupport.createPersistent(NODE_ROOT_PATH);
            } else {
                log.info("zookeeper root node '/rx-seat' has been created...");
            }

            // 创建此坐席服务节点的zk node根节点
            String serviceRootPath = getServiceRootPath();
            zookeeperSupport.createPersistent(serviceRootPath, SERVICE_UUID + "\t" + LocalDateTime.now().toString());

            // 监听 此服务节点的 子节点变化
            CuratorCacheListener cacheListener = CuratorCacheListener
                    .builder()
                    .forPathChildrenCache(serviceRootPath, curatorFramework,
                            new PathChildrenCacheListenerImpl())
                    .build();

            CuratorCache curatorCache = CuratorCache.builder(curatorFramework, serviceRootPath).build();
            curatorCache.listenable().addListener(cacheListener, zookeeperThreadPool);
            curatorCache.start();
        } catch (Exception e) {
            log.error("curator start exception", e);
            throw e;
        }
    }

    public class PathChildrenCacheListenerImpl implements PathChildrenCacheListener, ZkConst {

        public void handleEvent(String eventUserId, String eventId) {
            log.info("do eventId:{} userId:{}", eventId, eventUserId);
            UserHook userHook = SpringBeanSupport.getBean(UserHook.class);
            userHook.doEvent(eventId, eventUserId);
        }

        public void doService(String dataPath, String content) {
            if (content.startsWith(HANDLE_EVENT_PREFIX)) {
                handleEvent(getEventUserId(content), getEventId(content));
            }
        }

        @Override
        public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
            Type eventType = event.getType();
            ChildData childData = event.getData();

            if (childData == null) {
                log.info("data is null {}", event);
                return;
            }

            String dataPath = childData.getPath();
            String content = getContent(childData);
            log.info("type:{} path:{} data:{}", eventType, dataPath, content);
            if (eventType == CHILD_REMOVED) {
                return;
            }

            if (eventType == CHILD_ADDED) {
                doService(dataPath, content);
            }

            CuratorStartRunner.this.zookeeperSupport.deletePath(dataPath);
        }
    }

}
